package com.hex.ds.hdtp.core.app.data.contrast.service.impl;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.hex.ds.hdtp.core.app.common.enums.BasicTaskStatus;
import com.hex.ds.hdtp.core.app.common.util.DatabaseStreamContrastUtil;
import com.hex.ds.hdtp.core.app.data.contrast.converter.ICrossDbContrastConverter;
import com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastCommonDto;
import com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastDataSqlDto;
import com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastResultDto;
import com.hex.ds.hdtp.core.app.data.contrast.dto.CrossDbContrastSampleDataDto;
import com.hex.ds.hdtp.core.app.data.contrast.service.ICrossDbContrastRunLogService;
import com.hex.ds.hdtp.core.app.data.table.dto.request.TableTransferFieldMapRequestDto;
import com.hex.ds.hdtp.core.app.data.table.dto.response.TableTransferFieldMapResponseDto;
import com.hex.ds.hdtp.core.app.data.table.service.ITableTransferFieldMapService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceInfoService;
import com.hex.ds.hdtp.core.inf.common.utils.JdbcUtil;
import com.hex.ds.hdtp.core.inf.data.contrast.service.IDataInfoService;
import com.hex.ds.hdtp.core.inf.data.contrast.service.impl.DataInfoSelectHandlerContext;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.repository.ISourceConfInfoRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

import static com.hex.ds.hdtp.core.app.data.contrast.service.impl.ContrastSampleDataService.*;

/**
 * @Package com.hex.ds.hdtp.core.app.data.contrast.service.impl
 * @ClassName StreamReaderToContrastService
 * @Description 流式跨库比对-功能实现
 * @Author Wang zhihao
 * @Date 2023/8/6 22:07
 * @Version v1.0
 **/
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class StreamReaderToContrastService {
    volatile static int count = 0;
    private final ITableTransferFieldMapService tableTransferFieldMapService;
    private final DataInfoSelectHandlerContext dataInfoSelectHandlerContext;
    private final ISourceInfoService sourceInfoService;


    private final ICrossDbContrastRunLogService crossDbContrastRunLogService;

    private final ICrossDbContrastConverter contrastCommonConverter;

    private IDataInfoService getIDataInfoService(String type) {
        return dataInfoSelectHandlerContext.getDataInfoService(type);
    }

    private final ISourceConfInfoRepository sourceConfInfoRepository;
    private Map<String, Boolean> contrastResult = new HashMap<>();

    /**
     * @param contrastInfo
     * @param sqlDto
     * @Method streamContrast
     * @Return java.lang.Boolean
     * @Description 通过流式对比JDBC查询出的数据
     * @Author Wang zhihao
     * @Date 2023/8/6 21:55
     * @Version V1.0
     */
    public CrossDbContrastResultDto streamContrast(CrossDbContrastCommonDto contrastInfo, CrossDbContrastDataSqlDto sqlDto, CrossDbContrastSampleDataDto contrastSampleData, String currentTread, String runLogPkId) throws Exception {
        if (!ContrastSampleDataService.crossDbContrastResultDtoMap.get(currentTread).isContrastResult() && ContrastSampleDataService.crossDbContrastResultDtoMap.get(currentTread).getSampleDataList().size() >= contrastSampleData.getSampleDataCount())
            return crossDbContrastResultDtoMap.get(currentTread);
        if (crossDbContrastRunLogService.queryStatusByPkId(runLogPkId) != null
                && crossDbContrastRunLogService.queryStatusByPkId(runLogPkId).equalsIgnoreCase(BasicTaskStatus.STOPPED.name()))
            return null;
        List<SourceConfInfoPo> sourceConfInfoPoList = sourceConfInfoRepository.queryListBySourceCode(contrastInfo.getSourceCode());
        List<SourceConfInfoPo> contrastConfInfoPoList = sourceConfInfoRepository.queryListBySourceCode(contrastInfo.getContrastCode());
        //目标表精度
        List<Map<String, String>> accuracyList = getIDataInfoService(contrastConfInfoPoList.get(0).getSourceType()).selectFieldAccuracy(contrastInfo.getContrastCode(), contrastInfo.getContrastSchema(), contrastInfo.getContrastTable());
        // 去除 "column_accuracy" 值为 "0" 的 Map 条目
        accuracyList.removeIf(map -> "0".equals(map.get("column_accuracy")));
        // 构建新的Map
        Map<String, String> resultMap = accuracyList.stream().collect(Collectors.toMap(map -> map.get("column_name"), map -> map.get("column_accuracy")));
        //字段名映射
        List<TableTransferFieldMapResponseDto> tableTransferFieldMaps = new ArrayList<>();
        if (contrastInfo.getTableTransferPkId() != null) {
            //如果经历过表迁移，那么就从表迁移里获取字段映射
            TableTransferFieldMapRequestDto tableTransferFieldMapRequestDto = new TableTransferFieldMapRequestDto();
            tableTransferFieldMapRequestDto.setListPkId(contrastInfo.getTableTransferPkId());
            tableTransferFieldMaps = tableTransferFieldMapService.queryList(tableTransferFieldMapRequestDto);
        } else {
            //否则就连接表获取字段映射
            tableTransferFieldMaps = sourceInfoService.getFieldMaps(contrastCommonConverter.toBasicTransferInfo(contrastInfo));

        }
        Map<String, String> fieldsMap = tableTransferFieldMaps.stream()
                .collect(Collectors.toMap(TableTransferFieldMapResponseDto::getTargetFieldName, TableTransferFieldMapResponseDto::getSourceFieldName));
        // 创建目标Map<映射后字段，精度>
        Map<String, String> targetMap = new HashMap<>();
        // 遍历源Map的键值对
        for (Map.Entry<String, String> entry : resultMap.entrySet()) {
            String key = entry.getKey();
            String value = entry.getValue();
            // 如果键在映射关系Map中存在，则使用映射的新键，否则保持原键
            String newKey = fieldsMap.getOrDefault(key, key);
            // 将新键和值添加到目标Map中
            targetMap.put(newKey.toUpperCase(), value);
        }
        //查询sql 获取源和目标表的ResultSer
        ResultSet sourceData = queryData(sourceConfInfoPoList, sqlDto.getSourceQuerySql());
        ResultSet targetData = queryData(contrastConfInfoPoList, sqlDto.getTargetQuerySql());
        //查询结果 放入集合
        Queue sourceQueue = DatabaseStreamContrastUtil.dataToQueueConvertType(sourceData, contrastSampleData.getIsCharNullToEmpty(), contrastSampleData.getIsNumberNullToZero(), contrastInfo, targetMap);
        Map<Integer, Map<String, Object>> targetQueue = DatabaseStreamContrastUtil.dataToMap(targetData, contrastSampleData.getIsCharNullToEmpty(), contrastSampleData.getIsNumberNullToZero(), contrastInfo, targetMap);
        HashMap<String, Queue> sourceDataMap = new HashMap<>();
        HashMap<String, Map<Integer, Map<String, Object>>> targetDataMap = new HashMap<>();
        sourceDataMap.put(Thread.currentThread().getName(), sourceQueue);
        targetDataMap.put(Thread.currentThread().getName(), targetQueue);
        //比对逻辑
        crossDbContrastResultDtoMap.put(currentTread, dataContrast(sourceDataMap, targetDataMap, contrastSampleData, currentTread, runLogPkId));
        sourceDataMap.remove(Thread.currentThread().getName());
        targetDataMap.remove(Thread.currentThread().getName());
        return crossDbContrastResultDtoMap.get(currentTread);
    }


    /**
     * @param sourceConfInfoPoList
     * @param querySql
     * @Method queryData
     * @Param
     * @Return java.sql.ResultSet
     * @Description JDBC通过流的方式读取数据, 限制一次数据量为fetchSize, 超时时间为queryTimeOut
     * @Author Wang zhihao
     * @Date 2023/8/18 9:33
     * @Version V1.0
     */
    public ResultSet queryData(List<SourceConfInfoPo> sourceConfInfoPoList, String querySql) throws Exception {
        Connection connection = null;
        Statement statement = null;
        ResultSet resultSet = null;
        try {
            int fetchSize = 2500;
            int queryTimeOut = 3600;
            connection = JdbcUtil.getJdbcConnection(sourceConfInfoPoList);
            statement = JdbcUtil.createJdbcStatement(connection);
            statement.setFetchSize(fetchSize);
            // 如果驱动不是Hive.v1.1.0，设置SQL超时时间为queryTimeOut
            String driverClassName = connection.getMetaData().getClass().getName();
            if (!driverClassName.equals("com.hex.hdtp.shaded.hive.v110.jdbc.HiveDatabaseMetaData")) {
                statement.setQueryTimeout(queryTimeOut);
            }
            resultSet = JdbcUtil.executeQuerySql(statement, querySql);
        } catch (Exception e) {
            log.error(StrUtil.format("{}执行失败,Exception", querySql), e);
            log.error(Arrays.toString(e.getStackTrace()));
            throw new Exception(e);
        }
        return resultSet;
    }

    /**
     * @Method dataContrast
     * @Param
     * @Return boolean
     * @Description 多线程进行数据的对比
     * @Author Wang zhihao
     * @Date 2023/8/6 21:56
     * @Version V1.0
     */
    public CrossDbContrastResultDto dataContrast(HashMap<String, Queue> sourceDataMap, HashMap<String, Map<Integer, Map<String, Object>>> targetDataMap, CrossDbContrastSampleDataDto contrastSampleData, String currentTread, String runLogPkId) throws InterruptedException {
        if (crossDbContrastRunLogService.queryStatusByPkId(runLogPkId) != null
                && crossDbContrastRunLogService.queryStatusByPkId(runLogPkId).equalsIgnoreCase(BasicTaskStatus.STOPPED.name())) {
            log.info("检测到{}比对任务停止", currentTread);
            return null;
        }
        int count = 0;
        Queue<Map<Integer, Map<String, Object>>> queue = sourceDataMap.get(Thread.currentThread().getName());
        Map<Integer, Map<String, Object>> map = targetDataMap.get(Thread.currentThread().getName());
        while (!queue.isEmpty()) {
            if (!ContrastSampleDataService.crossDbContrastResultDtoMap.get(currentTread).isContrastResult() && ContrastSampleDataService.crossDbContrastResultDtoMap.get(currentTread).getSampleDataList().size() >= contrastSampleData.getSampleDataCount())
                break;
            // Queue.poll()可以在取数据后移除队列
            Map<Integer, Map<String, Object>> result = queue.poll();
            // 队列中数据比对
            count = contrastQueueData(contrastSampleData, currentTread, result, map, count);
        }
        map.clear();
        queue.clear();
        return ContrastSampleDataService.crossDbContrastResultDtoMap.get(currentTread).setSampleData(JSONObject.toJSONString(ContrastSampleDataService.crossDbContrastResultDtoMap.get(currentTread).getSampleDataList(), SerializerFeature.WriteMapNullValue));
    }

    /**
     * @Method contrastQueueData
     * @Param contrastSampleData
     * @Param currentTread
     * @Param result
     * @Param map
     * @Param count
     * @Return int
     * @Description 队列中数据比对
     * @Author gj.xu
     * @Date 2024/2/22 18:42
     * @Version V2.0
     */
    private int contrastQueueData(CrossDbContrastSampleDataDto contrastSampleData, String currentTread, Map<Integer, Map<String, Object>> result, Map<Integer, Map<String, Object>> map, int count) {
        Set<Integer> keySet = result.keySet();
        for (Integer key : keySet) {
            List<Map<String, Object>> queueList = new ArrayList<>();
            List<Map<String, Object>> mapList = new ArrayList<>();
            queueList.add(result.get(key));
            mapList.add(map.get(key));
            contrastResult.put(currentTread, CollectionUtils.isEqualCollection(queueList, mapList));
            queueList.clear();
            mapList.clear();
            //DCL提高性能
            synchronized (this) {
                //volatile修饰，防止JIT优化
                if (!contrastResult.get(currentTread)) {
                    checkContrastFlag(contrastSampleData, currentTread);
                    // 异常数据抽取
                    count++;
                    if (isFillErrorSampleData(contrastSampleData, currentTread, count)) {
                        fillErrorSampleData(contrastSampleData, currentTread, result, map, key);
                    } else break;
                }
            }
            map.remove(key);
        }
        return count;
    }

    private void checkContrastFlag(CrossDbContrastSampleDataDto contrastSampleData, String currentTread) {
        if (currentIsBatch.get(currentTread).equals(Boolean.FALSE))
            crossDbContrastResultDtoMap.get(currentTread).setContrastResult(false);
            //如果是分批策略的话状态为逻辑失败，等线程池任务全部完成就会真正失败
        else if (crossDbContrastResultDtoMap.get(currentTread).getSampleDataList().size() >= contrastSampleData.getSampleDataCount())
            currentFlag.put(currentTread, false);
    }

    /**
     * @Method fillErrorSampleData
     * @Param contrastSampleData
     * @Param currentTread
     * @Param result
     * @Param map
     * @Param key
     * @Return void
     * @Description 抽取异常数据
     * @Author gj.xu
     * @Date 2024/2/22 19:37
     * @Version V2.0
     */
    private void fillErrorSampleData(CrossDbContrastSampleDataDto contrastSampleData, String currentTread, Map<Integer, Map<String, Object>> result, Map<Integer, Map<String, Object>> map, Integer key) {
        sampleData(result.get(key), map.get(key), crossDbContrastResultDtoMap.get(currentTread));
        if (currentIsBatch.get(currentTread).equals(Boolean.TRUE)
                && crossDbContrastResultDtoMap.get(currentTread).getSampleDataList().size() >= contrastSampleData.getSampleDataCount())
            currentFlag.put(currentTread, false);
    }

    /**
     * @Method isFillErrorSampleData
     * @Param contrastSampleData
     * @Param currentTread
     * @Return java.lang.Boolean
     * @Description 判断是否要抽取异常数据
     * @Author gj.xu
     * @Date 2024/2/22 19:10
     * @Version V2.0
     */
    private Boolean isFillErrorSampleData(CrossDbContrastSampleDataDto contrastSampleData, String currentTread, int count) {
        return (count < contrastSampleData.getSampleDataCount()
                && currentIsBatch.get(currentTread)
                && StrUtil.equalsAnyIgnoreCase("Y", contrastSampleData.getIsSampleData())
        ) || (StrUtil.equalsAnyIgnoreCase("Y", contrastSampleData.getIsSampleData())
                && crossDbContrastResultDtoMap.get(currentTread).getSampleDataList().size() < contrastSampleData.getSampleDataCount());
    }

    @Async("dataDetailContrastAsyncTaskExecutor")
    public CompletableFuture<CrossDbContrastResultDto> sampleContrast(CrossDbContrastCommonDto contrastCommon, CrossDbContrastSampleDataDto contrastSampleData, long i, String currentTread, String runLogPkId) {
        CompletableFuture<CrossDbContrastResultDto> crossDbContrastResultDtoCompletableFuture = new CompletableFuture<>();
        if (crossDbContrastRunLogService.queryStatusByPkId(runLogPkId) != null
                && crossDbContrastRunLogService.queryStatusByPkId(runLogPkId).equalsIgnoreCase(BasicTaskStatus.STOPPED.name()))
            return null;
        if (!crossDbContrastResultDtoMap.get(currentTread).isContrastResult() && crossDbContrastResultDtoMap.get(currentTread).getSampleDataList().size() >= contrastSampleData.getSampleDataCount()) {
            crossDbContrastResultDtoCompletableFuture.complete(ContrastSampleDataService.crossDbContrastResultDtoMap.get(currentTread));
            return crossDbContrastResultDtoCompletableFuture;
        }
        log.info("触发{}的第{}批异步比对任务", currentTread, i + 1);
        try {
            //比对逻辑
            CrossDbContrastResultDto crossDbContrastResultDto = streamContrast(contrastCommon, sqlDtoMap.get(currentTread).get(i), contrastSampleData, currentTread, runLogPkId);
            crossDbContrastResultDtoCompletableFuture.complete(crossDbContrastResultDto);
            if (!currentFlag.get(currentTread)) {
                Integer cnt = currentCount.get(currentTread);
                //错误信息收集已加锁，可以忽略currentCount多减
                currentCount.put(currentTread, --cnt);
            }
            if (currentCount.get(currentTread) <= 0)
                crossDbContrastResultDtoMap.get(currentTread).setContrastResult(false);
            return crossDbContrastResultDtoCompletableFuture;
        } catch (Exception e) {
            log.error(StrUtil.format("{}的第{}批异步比对失败", currentTread, i + 1), e);
            crossDbContrastResultDtoMap.get(currentTread).setContrastResult(false);
            crossDbContrastResultDtoMap.get(currentTread).setSampleData(e.getMessage());
            throw new RuntimeException(e);
        } finally {
            log.info("{}的第{}批异步比对任务结束", currentTread, i + 1);
            sqlDtoMap.get(currentTread).remove(i);
            countDownLatchMap.get(currentTread).countDown();
            if (!ContrastSampleDataService.crossDbContrastResultDtoMap.get(currentTread).isContrastResult() && ContrastSampleDataService.crossDbContrastResultDtoMap.get(currentTread).getSampleDataList().size() >= contrastSampleData.getSampleDataCount()) {
                while (countDownLatchMap.get(currentTread).getCount() != 0) {
                    countDownLatchMap.get(currentTread).countDown();
                }
            }
        }
    }

    public void sampleData(Map<String, Object> sourceMap, Map<String, Object> targetMap, CrossDbContrastResultDto resultDto) {
        ArrayList<String> fields = new ArrayList<>();
        CrossDbContrastResultDto.SampleData sampleData = new CrossDbContrastResultDto.SampleData();
        Iterator<String> iterator = sourceMap.keySet().iterator();
        while (iterator.hasNext()) {
            String next = iterator.next();
            Object sourceData = sourceMap.get(next);
            Object targetData = targetMap.get(next);

            if (sourceData != null && !sourceData.equals(targetData)) {
                fields.add(next);
            } else if (sourceData == null && targetData != null) {
                fields.add(next);
            }
        }
        sampleData.setFields(fields).setSource(sourceMap).setTarget(targetMap);
        resultDto.getSampleDataList().add(sampleData);
    }

    @Async("dataContrastAsyncTaskExecutor")
    public boolean doDataContrast(LinkedList<String> sourceResultList, LinkedList<String> targetResultList) {
        //行数据量是否一致
        if (sourceResultList.size() != targetResultList.size()) {
            return false;
        }
        Iterator<String> sourceIterator = sourceResultList.iterator();
        Iterator<String> targetIterator = targetResultList.iterator();
        // 遍历链表
        while (sourceIterator.hasNext() && targetIterator.hasNext()) {
            String sourceLine = sourceIterator.next();
            String targetLine = targetIterator.next();
            // 比较对应行的数据是否一致
            if (!sourceLine.equals(targetLine)) {
                return false;
            }
        }
        return true;
    }

}
